package cn.classg.rabbitmq.handle;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.DeclarableCustomizer;
import org.springframework.amqp.core.Declarables;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * @author CLASSG
 */
@Slf4j
public class RabbitAdminDeclaresHandle implements InitializingBean,ApplicationContextAware {

    private ApplicationContext applicationContext;

    private static final String DELAYED_MESSAGE_EXCHANGE = "x-delayed-message";

    /**
     * The default exchange name.
     */
    public static final String DEFAULT_EXCHANGE_NAME = "";

    private RabbitTemplate rabbitTemplate;

    public RabbitAdminDeclaresHandle(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (this.applicationContext == null) {
            log.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings");
            return;
        }
        log.debug("Initializing declarations");
        Collection<Exchange> contextExchanges = new LinkedList<Exchange>(
                this.applicationContext.getBeansOfType(Exchange.class).values());
        Collection<Queue> contextQueues = new LinkedList<Queue>(
                this.applicationContext.getBeansOfType(Queue.class).values());
        Collection<Binding> contextBindings = new LinkedList<Binding>(
                this.applicationContext.getBeansOfType(Binding.class).values());
        Collection<DeclarableCustomizer> customizers =
                this.applicationContext.getBeansOfType(DeclarableCustomizer.class).values();

        processDeclarables(contextExchanges, contextQueues, contextBindings);

        final Collection<Exchange> exchanges = filterDeclarables(contextExchanges, customizers);
        final Collection<Queue> queues = filterDeclarables(contextQueues, customizers);
        final Collection<Binding> bindings = filterDeclarables(contextBindings, customizers);

        for (Exchange exchange : exchanges) {
            if ((!exchange.isDurable() || exchange.isAutoDelete())  && log.isInfoEnabled()) {
                log.info("Auto-declaring a non-durable or auto-delete Exchange ("
                        + exchange.getName()
                        + ") durable:" + exchange.isDurable() + ", auto-delete:" + exchange.isAutoDelete() + ". "
                        + "It will be deleted by the broker if it shuts down, and can be redeclared by closing and "
                        + "reopening the connection.");
            }
        }

        for (Queue queue : queues) {
            if ((!queue.isDurable() || queue.isAutoDelete() || queue.isExclusive()) && log.isInfoEnabled()) {
                log.info("Auto-declaring a non-durable, auto-delete, or exclusive Queue ("
                        + queue.getName()
                        + ") durable:" + queue.isDurable() + ", auto-delete:" + queue.isAutoDelete() + ", exclusive:"
                        + queue.isExclusive() + ". "
                        + "It will be redeclared if the broker stops and is restarted while the connection factory is "
                        + "alive, but all messages will be lost.");
            }
        }

        if (exchanges.size() == 0 && queues.size() == 0 && bindings.size() == 0) {
            log.debug("Nothing to declare");
            return;
        }
        this.rabbitTemplate.execute(channel -> {
            declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
            declareQueues(channel, queues.toArray(new Queue[queues.size()]));
            declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
            return null;
        });
    }

    private <T extends Declarable> Collection<T> filterDeclarables(Collection<T> declarables,
                                                                   Collection<DeclarableCustomizer> customizers) {

        return declarables.stream()
                .filter(dec -> dec.shouldDeclare())
                .map(dec -> {
                    if (customizers.isEmpty()) {
                        return dec;
                    }
                    AtomicReference<T> ref = new AtomicReference<>(dec);
                    customizers.forEach(cust -> ref.set((T) cust.apply(ref.get())));
                    return ref.get();
                })
                .collect(Collectors.toList());
    }


    private void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,
                                    Collection<Binding> contextBindings) {

        Collection<Declarables> declarables = this.applicationContext.getBeansOfType(Declarables.class, false, true)
                .values();
        declarables.forEach(d -> {
            d.getDeclarables().forEach(declarable -> {
                if (declarable instanceof Exchange) {
                    contextExchanges.add((Exchange) declarable);
                }
                else if (declarable instanceof Queue) {
                    contextQueues.add((Queue) declarable);
                }
                else if (declarable instanceof Binding) {
                    contextBindings.add((Binding) declarable);
                }
            });
        });
    }

    private void declareBindings(final Channel channel, final Binding... bindings) throws IOException {
        for (Binding binding : bindings) {
            if (log.isDebugEnabled()) {
                log.debug("Binding destination [" + binding.getDestination() + " (" + binding.getDestinationType()
                        + ")] to exchange [" + binding.getExchange() + "] with routing key [" + binding.getRoutingKey()
                        + "]");
            }

            try {
                if (binding.isDestinationQueue()) {
                    if (!isDeclaringImplicitQueueBinding(binding)) {
                        channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
                                binding.getArguments());
                    }
                }
                else {
                    channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
                            binding.getArguments());
                }
            }
            catch (IOException e) {
                log.warn("Failed to declare " + binding
                        + ": " + (binding == null ? "broker-generated" : binding)
                        + ", continuing... " + e);
            }
        }
    }

    private boolean isDeclaringImplicitQueueBinding(Binding binding) {
        if (isImplicitQueueBinding(binding)) {
            log.debug("The default exchange is implicitly bound to every queue," +
                    " with a routing key equal to the queue name.");
            return true;
        }
        return false;
    }

    private boolean isImplicitQueueBinding(Binding binding) {
        return isDefaultExchange(binding.getExchange()) && binding.getDestination().equals(binding.getRoutingKey());
    }

    private void declareExchanges(final Channel channel, final Exchange... exchanges) throws IOException {
        for (final Exchange exchange : exchanges) {
            if (log.isDebugEnabled()) {
                log.debug("declaring Exchange '" + exchange.getName() + "'");
            }

            if (!isDeclaringDefaultExchange(exchange)) {
                try {
                    if (exchange.isDelayed()) {
                        Map<String, Object> arguments = exchange.getArguments();
                        if (arguments == null) {
                            arguments = new HashMap<String, Object>();
                        }
                        else {
                            arguments = new HashMap<String, Object>(arguments);
                        }
                        arguments.put("x-delayed-type", exchange.getType());
                        channel.exchangeDeclare(exchange.getName(), DELAYED_MESSAGE_EXCHANGE, exchange.isDurable(),
                                exchange.isAutoDelete(), exchange.isInternal(), arguments);
                    }
                    else {
                        channel.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(),
                                exchange.isAutoDelete(), exchange.isInternal(), exchange.getArguments());
                    }
                }
                catch (IOException e) {
                    log.warn("Failed to declare " + exchange
                            + ": " + (exchange == null ? "broker-generated" : exchange)
                            + ", continuing... " + e);
                }
            }
        }
    }

    private AMQP.Queue.DeclareOk[] declareQueues(final Channel channel, final Queue... queues) throws IOException {
        List<AMQP.Queue.DeclareOk> declareOks = new ArrayList<AMQP.Queue.DeclareOk>(queues.length);
        for (int i = 0; i < queues.length; i++) {
            Queue queue = queues[i];
            if (!queue.getName().startsWith("amq.")) {
                if (log.isDebugEnabled()) {
                    log.debug("declaring Queue '" + queue.getName() + "'");
                }
                try {
                    try {
                        AMQP.Queue.DeclareOk declareOk = channel.queueDeclare(queue.getName(), queue.isDurable(),
                                queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
                        if (StringUtils.hasText(declareOk.getQueue())) {
                            queue.setActualName(declareOk.getQueue());
                        }
                        declareOks.add(declareOk);
                    }
                    catch (IllegalArgumentException e) {
                        closeChannelAfterIllegalArg(channel, queue);
                        throw new IOException(e);
                    }
                }
                catch (IOException e) { // NOSONAR exceptions for flow control
                    log.warn("Failed to declare " + queue
                            + ": " + (queue == null ? "broker-generated" : queue)
                            + ", continuing... " + e);
                }
            }
            else if (log.isDebugEnabled()) {
                log.debug(queue.getName() + ": Queue with name that starts with 'amq.' cannot be declared.");
            }
        }
        return declareOks.toArray(new AMQP.Queue.DeclareOk[declareOks.size()]);
    }

    private boolean isDeclaringDefaultExchange(Exchange exchange) {
        if (isDefaultExchange(exchange.getName())) {
            log.debug("Default exchange is pre-declared by server.");
            return true;
        }
        return false;
    }

    private boolean isDefaultExchange(String exchangeName) {
        return DEFAULT_EXCHANGE_NAME.equals(exchangeName);
    }

    private void closeChannelAfterIllegalArg(final Channel channel, Queue queue) {
        if (log.isDebugEnabled()) {
            log.error("Exception while declaring queue: '" + queue.getName() + "'");
        }
        try {
            if (channel instanceof ChannelProxy) {
                ((ChannelProxy) channel).getTargetChannel().close();
            }
        }
        catch (IOException | TimeoutException e1) {
            log.error("Failed to close channel after illegal argument", e1);
        }
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
    }
}
